
#include <mpi.h>
#include <string.h>
#include <sys/wait.h>
#include <set>
#include "parser.h"
#include "workStealing.h"

/*
 * Global variables, used in all functions
 */
vector<Node*>	nodes;	// All nodes of the graph
list<Node*>	tasks_ready; // Tasks ready to run for this machine
list<Node*>	tasks_not_ready;	// Tasks which are waiting some dependencies, for this machine
int		machine_id	= -1;	// Our machine id
int		nb_machines	= 0;
int		root_node_id;
bool	all_tasks_ended	= false;
bool	print_graph	= false;
vector<Node*>	nodes_ready;	// Nodes current machine can work on
vector<Node*>	nodes_blocked; //  Nodes current machine is waiting for
pthread_mutex_t	nodes_ready_mutex;
pthread_mutex_t	nodes_blocked_mutex;
pthread_t	communication_dispatcher_thread;
pthread_t	work_stealing_undergone_thread;
pthread_t	task_end_thread;
pthread_t	last_node_thread;
clock_t		time_init;

/*
 * Program finalize, to call before leaving
 */
void finalize()
{
	//pthread_join(work_stealing_thread, NULL);
	//pthread_join(work_stealing_undergone_thread, NULL);
	//pthread_join(communication_dispatcher_thread, NULL);
	//pthread_cancel(communication_dispatcher_thread);
	pthread_cancel(work_stealing_undergone_thread);
	pthread_cancel(task_end_thread);
	pthread_cancel(last_node_thread);
	MPI_Finalize();
}
/*
 * Program init
 */
void init(int argc, char **argv)
{
	//time_init = time(NULL);
	//time_init = clock();
	Node *root = NULL;
	// Find root node
	if(argc>1) { // Trying to make a specific target
#ifdef DEBUG
		cout << time(NULL)<<" " << "Making specific target " << argv[1] << endl;
#endif
		if(strncmp(argv[1],"--printgraph",13)==0) {
			print_graph = true;
			root = nodes[0];
		}
		for(vector<Node*>::iterator it=nodes.begin();it!=nodes.end();++it) {
			if(!strcmp((*it)->name.c_str(),argv[1])) {
				root	= *it;
			}
		}
	} else { 	// Trying to make the first target
		root	= nodes[0];
	}
	if(NULL==root) { // No root node found
		cerr << "No Makefile target ";
		if(argc>1) {
			cerr << "("<< argv[1] << ") ";
		}
		cerr << "found" << endl;
		exit(1);
	}
	if(argc>2) {
		if(strncmp(argv[1],"--printgraph",13)==0) {
			print_graph = true;
		}
	}
	// MPI init
	//MPI_Init(&argc,&argv);
	int provided;
	MPI_Init_thread(&argc, &argv, MPI_THREAD_MULTIPLE, &provided);
	if(provided!=MPI_THREAD_MULTIPLE) {
		cout << "ERROR: MPI was not able to use MPI_THREAD_MULTIPLE" << endl;
		exit(1);
	}
	//machine_id	= 0; // TODO: use MPI rank instead of static value
	MPI_Comm_rank(MPI_COMM_WORLD,&machine_id);
	//nb_machines	= 1; // TODO: use MPI number of process instead of static value
	MPI_Comm_size(MPI_COMM_WORLD,&nb_machines);

	// Threads init
	//pthread_create(&work_stealing_thread, NULL, work_stealing_thread_function, NULL);
	pthread_create(&work_stealing_undergone_thread, NULL, work_stealing_undergone_thread_function, NULL);
	pthread_create(&task_end_thread, NULL, task_end_thread_function, NULL);
	pthread_create(&last_node_thread, NULL, last_node_thread_function, NULL);
	
	//pthread_create(&communication_dispatcher_thread, NULL, communication_dispatcher_function, NULL);
	/*int ret_fork = fork();
	cout << "PID: " << ret_fork << endl;
	if(ret_fork) {
		//communication_dispatcher_function(NULL);
		sleep(2);
		char request_tag;
		MPI_Recv(&request_tag, 1, MPI_CHAR, MPI_ANY_SOURCE, TAG_REQUEST, MPI_COMM_WORLD, NULL);
		exit(0);
	}*/
	// Mutex init
	pthread_mutex_init(&nodes_ready_mutex, NULL);
	pthread_mutex_init(&nodes_blocked_mutex, NULL);

	// graph init
	graph_tasks_affectation_init(root, machine_id, nodes.size());
	root_node_id = root->id;
	if(print_graph) {
		if(machine_id==0) { // Only the master process draw the graph
			printGraph(nodes);
		}
		finalize();
		exit(0);
	}
}

void task_ended_message_received(int task_dep_id,int task_ended_id)
{
	Node *node	= nodes[task_ended_id];
	Node *node_dep	= nodes[task_dep_id];
	// Find node in dependencies
	for(list<Node*>::iterator it=node_dep->listDependencies.begin();it!=node_dep->listDependencies.end();++it) {
		if(*it==node) {
			node_dep->listDependencies.erase(it);
			cout << time(NULL)<<" " << machine_id << " Removing node " << *node << "(" << node_dep->listDependencies.size() << " left)" << endl;
			break;
		}
	}
	if(node_dep->listDependencies.empty()) { // No more dependencies for this node
		cout << time(NULL)<< " " << machine_id << " Adding node " << *node_dep << " to nodes_ready" << endl;
		// search
		for(vector<Node*>::iterator it_search=nodes_blocked.begin();it_search!=nodes_blocked.end();++it_search) {
			if(node_dep==*it_search) {
				all_tasks_ended=false;
				nodes_ready.push_back(node_dep);
				nodes_blocked.erase(it_search);
				break;
			}
		}
	}
}

void execute_task(Node *node)
{
	// Executing node
	cout << time(NULL)<<" " << machine_id << " Executing node " << *node << "("<< nodes_ready.size() << ")" <<endl;
	// if it is the last node (root), prevent all nodes it's over now !
	if(node->id==root_node_id) {
		all_tasks_ended	= true;
		for(int machine_to_prevent=0;machine_to_prevent<nb_machines;machine_to_prevent++) {
			if(machine_to_prevent!=machine_id) {
				//sleep(2);
				cout << time(NULL)<<" " << machine_id << " THIS IS THE END message sent to" << machine_to_prevent << endl;
				/*char request = TAG_NODE_END;
				  print_MPI_error(MPI_Send(&request, 1, MPI_CHAR, machine_to_prevent, TAG_REQUEST, MPI_COMM_WORLD));
				  print_MPI_error(MPI_Send(&(node->id), 1, MPI_INT, machine_to_prevent, TAG_NODE_END, MPI_COMM_WORLD));*/
				char request = TAG_LAST_TASK_END;
				MPI_Request r3;
				print_MPI_error(MPI_Isend(&request, 1, MPI_CHAR, machine_to_prevent, TAG_REQUEST, MPI_COMM_WORLD, &r3));
				//sleep(2);
				char tmp=1;
				MPI_Request r4;
				print_MPI_error(MPI_Isend(&tmp, 1, MPI_CHAR, machine_to_prevent, TAG_LAST_TASK_END, MPI_COMM_WORLD, &r4));
				//sleep(2);
			}
		}
	}
	for(vector<string>::iterator it_command=node->commands.begin();it_command!=node->commands.end();++it_command) {
		cout << time(NULL)<<" " << (*it_command) << endl;
		size_t command_size = strlen((*it_command).c_str());
		char* command = (char*)malloc(sizeof(char)*(command_size+1));
		strncpy(command, (*it_command).c_str(), command_size+1);
		//cout << command << endl;
		//system((*it_command).c_str());
		//signal(SIGCHLD, SIG_DFL);
		//alarm(0);
		//ualarm(0,0);
		//sleep(4);
		system(command);
		//alarm(500000);
		//ualarm(500000,500000);
		/*int ret_fork = fork();
		  if (!ret_fork) { // if it's the new process running, exec command
		  fflush(stdout);
		  system(command);
		//execvp(cmd[0], &cmd[0]);
		exit(-1); // this instruction is supposed to never be executed
		&	}
		cout << "Waiting for " << command << endl;
		waitpid(ret_fork,0,0);*/
	}
	cout << time(NULL)<<" " << machine_id << " node " << *node << " done" << endl;

	// Send message to machine which have tasks waiting for task we are executing
	// TODO: do it using communication object
	set<int> set_machine_to_prevent;
	for(list<Node*>::iterator it=node->listDependant.begin();it!=node->listDependant.end();++it) {
		Node *node_dependant = *it;
		if(node_dependant->id_machine==machine_id) {
			task_ended_message_received(node_dependant->id, node->id);
		} else {
			set_machine_to_prevent.insert(node_dependant->id_machine);
		}
	}
	// send MPI message
	for(set<int>::iterator it_set=set_machine_to_prevent.begin();it_set!=set_machine_to_prevent.end();++it_set) {
		int machine_to_prevent_id = *it_set;
		cout << time(NULL)<<" " << machine_id <<  " Sending message to " << machine_to_prevent_id << " : " << node->id << " is ended " << endl;
		char request = TAG_NODE_END;
		MPI_Request r1;
		print_MPI_error(MPI_Isend(&request, 1, MPI_CHAR, machine_to_prevent_id, TAG_REQUEST, MPI_COMM_WORLD, &r1));
		MPI_Request r2;
		print_MPI_error(MPI_Isend(&(node->id), 1, MPI_INT, machine_to_prevent_id, TAG_NODE_END, MPI_COMM_WORLD, &r2));
	}
}

/*
 * Program main
 */
int main(int argc, char **argv)
{

	// Makefile parsing and graph construction
	nodes = parse();
#ifdef DEBUG
	for(int i=0; i<nodes.size(); i++) {
		cout << time(NULL)<<" " << *(nodes[i]) << "(" << nodes[i]->id << ")" <<endl;
	}
#endif
	// Initialisation and graph decoration (machines/tasks affectation)
	init(argc,argv);

	// Work:
	// while (hasTasks)
	while(!all_tasks_ended) {
	//while(!all_tasks_ended && (!nodes_ready.empty() || !nodes_blocked.empty())) {
		//while(!nodes_ready.empty() || !nodes_blocked.empty()) {
		while(nodes_ready.empty() && !all_tasks_ended) {
			// steal
			/*for(int i=0;i<nodes_blocked.size();i++){
			  cout << time(NULL)<<" " << *(nodes_blocked[i]) << endl;
			  }*/
			//sleep(1);
			work_stealing();
			if(nodes_ready.empty()) { // nothing stolen
				usleep(2000000);
			}
			/*MPI_Comm_rank(MPI_COMM_WORLD,&machine_id);
			  cout << time(NULL)<<" " << machine_id << endl;*/

		}
		if(!all_tasks_ended) {
			// Work
			// Take a 'ready to run' task
			pthread_mutex_lock(&nodes_ready_mutex);
			Node *node	= nodes_ready.back();
			nodes_ready.pop_back();
			pthread_mutex_unlock(&nodes_ready_mutex);
			// Execute it
			execute_task(node);
		}
	}
	// End
	cout << time(NULL) << " " << machine_id << "Process is going to leave: no more work to do" << endl;
	finalize();
	//cout << "Time: " << ((double)(clock()-time_init))/ CLOCKS_PER_SEC * 1000<<endl;

	return 0;
	}
